Add tunables bounding compaction pipeline memory - #1993
Conversation
The compaction read+write pipeline had fixed memory characteristics: the scanner decodes read.parquet.batch-size rows per Arrow batch (131072 by default) with no per-scan override, and each rolling data writer buffers up to 64 record batches in its input channel with no way to lower or raise that bound. On wide rows the product of those two numbers dominates a compaction worker's peak memory. - WithArrowBatchSize scan option caps the rows decoded per Arrow batch for one scan, overriding the table's read.parquet.batch-size without mutating caller-supplied scan options. - The rolling data writers' record channel capacity (default 64 batches) is now configurable via the WithRecordBatchBufferSize write option, threaded through recordWritingArgs to the writer factory. It applies to both the fanout and the clustered write paths, which share the rolling writer. - WithParquetRowGroupLimit write option overrides the table's write.parquet.row-group-limit per write, bounding rows buffered per row group before each flush. - Compaction group options forward all three into ExecuteCompactionGroup: WithCompactionReadBatchSize, WithCompactionRecordBatchBufferSize, and WithCompactionParquetRowGroupLimit. Signed-off-by: KranzL <50032317+KranzL@users.noreply.github.com>
zeroshade
left a comment
zeroshade
left a comment
There was a problem hiding this comment.
WithArrowBatchSize and WithParquetRowGroupLimit are correctly wired and pinned by tests, but WithRecordBatchBufferSize and all three ExecuteCompactionGroup forwardings can be deleted outright with the entire table package still passing.
Re-review verification: 0 of 1 prior findings confirmed fixed at f718b85 (each verified by mutating the fix and observing the suite go red, not by taking the claim on trust).
Verification performed
go build ./... (OK); go vet ./table/ ./catalog/hive/ (clean); go test ./table/ -count=1 -timeout=900s (ok, 8.1s); go test ./table/ -count=1 -race -timeout=900s with recordQueueCapacity() forced to return 1 (no deadlock/hang/race; only the deliberately-broken TestRecordQueueCapacityDefaultAndOverride assertions failed); targeted -v run of all 7 new tests (all PASS); 5 separate mutation experiments on arrow_scanner.go, write_records.go, rolling_data_writer.go (x2) and rewrite_data_files.go (x2); 2 throwaway probes in table/pr1993_probe_test.go (deleted). Final `git status --porcelain` empty.
This review was drafted by an AI-assisted tool and confirmed by an Apache Iceberg Go maintainer. After you've addressed the points above and pushed an update, an Apache Iceberg Go maintainer — a real person — will take the next look at the PR. The findings cite the project's review criteria; if you think one of them is mis-applied, please reply on the PR and a maintainer will weigh in.
More on how Apache Iceberg Go handles maintainer review: CONTRIBUTING.md.
| } | ||
|
|
||
| func TestRecordQueueCapacityDefaultAndOverride(t *testing.T) { | ||
| f := &writerFactory{} |
There was a problem hiding this comment.
major — WithRecordBatchBufferSize is unpinned end-to-end; its only test is a getter tautology
TestRecordQueueCapacityDefaultAndOverride constructs a bare &writerFactory{}, assigns .recordBufferSize directly, and asserts recordQueueCapacity() echoes it back. It never calls WithRecordBatchBufferSize, never populates recordWritingArgs, and never observes cap(recordCh) -- so the entire chain writeRecordConfig -> recordWritingArgs -> writerFactory -> channel capacity is untested. I verified the production wiring is in fact correct via a probe (cap 64 with default, cap 3 with override), so this is a test gap rather than a code defect, but the PR's headline knob currently has no regression protection. Suggest an end-to-end assertion on cap(w.recordCh) after newWriterFactory(recordWritingArgs{recordBatchBufferSize: N}, ...), mirroring the harness at table/partitioned_fanout_writer_test.go:175.
Evidence
Mutation 1 - replaced `recordBufferSize: args.recordBatchBufferSize,` with `recordBufferSize: 0,` at table/rolling_data_writer.go:223; `go test ./table/ -count=1 -timeout=900s` => `ok github.com/apache/iceberg-go/table 8.109s`. Mutation 2 (independent) - replaced `make(chan arrow.RecordBatch, w.recordQueueCapacity())` with `make(chan arrow.RecordBatch, rollingDataWriterQueueCapacity)` at table/rolling_data_writer.go:380; `go test ./table/ -count=1 -timeout=900s` => `ok github.com/apache/iceberg-go/table 5.922s`. Probe confirming the wiring itself works: `default (0) -> cap(recordCh)=64` / `override (3) -> cap(recordCh)=3`.
| } | ||
| if cfg.readBatchSize > 0 { | ||
| scanOpts = append(scanOpts, WithArrowBatchSize(cfg.readBatchSize)) | ||
| } |
There was a problem hiding this comment.
major — All three new ExecuteCompactionGroup forwardings can be deleted with the package still green, unlike the existing targetFileSize forwarding
The three new if cfg.X > 0 { ...append... } blocks that forward readBatchSize/recordBatchBufferSize/parquetRowGroupLimit into the scan and write options are covered only by TestCompactionGroupTuningOptions (write_read_tuning_test.go:120), which applies the CompactionGroupOption closures to a bare compactionGroupConfig struct and asserts the fields were set. It never calls ExecuteCompactionGroup, so nothing detects the forwarding being dropped. This is a deviation from the project's own convention: the pre-existing WithTargetFileSize forwarding IS pinned end-to-end. Suggest one ExecuteCompactionGroup test asserting output row groups are bounded by WithCompactionParquetRowGroupLimit, which is the cheapest observable of the three.
Evidence
Mutation - deleted the readBatchSize, recordBatchBufferSize and parquetRowGroupLimit append blocks from ExecuteCompactionGroup; `go test ./table/ -count=1 -timeout=900s` => `ok github.com/apache/iceberg-go/table 5.429s`. Control mutation - deleted only the PRE-EXISTING `if cfg.targetFileSize > 0 { writeOpts = append(writeOpts, WithTargetFileSize(cfg.targetFileSize)) }` block; same command => `FAIL github.com/apache/iceberg-go/table 5.355s`. Existing forwarding is pinned, new forwarding is not.
| // consumer buffers batches (e.g. a compaction's read+write pipeline). | ||
| // A non-positive value is ignored. | ||
| func WithArrowBatchSize(n int64) ScanOption { | ||
| if n <= 0 { |
There was a problem hiding this comment.
minor — WithArrowBatchSize is silently discarded when WithOptions is applied after it
WithArrowBatchSize stores into scan.options, but WithOptions (table/table.go:1280) does scan.options = maps.Clone(opts), replacing the map wholesale. WithArrowBatchSize is the first ScanOption to write into scan.options, so this ordering hazard is newly introduced by this PR. A caller who writes tbl.Scan(WithArrowBatchSize(n), WithOptions(userProps)) gets no cap and no error -- the memory bound silently does not apply. TestWithArrowBatchSizeDoesNotMutateCallerOptions only covers the working order (WithOptions first). Suggest either documenting the ordering requirement on WithArrowBatchSize, having WithOptions merge rather than replace, or storing the batch size in a dedicated Scan field like concurrency/limit rather than in the generic options map.
| // batch. A non-positive value keeps the table's | ||
| // read.parquet.batch-size property. | ||
| func WithCompactionReadBatchSize(n int64) CompactionGroupOption { | ||
| return func(c *compactionGroupConfig) { |
There was a problem hiding this comment.
minor — WithCompactionReadBatchSize doc overstates the bound it provides
The comment claims the two options together bound "the memory held by the compaction's read+write pipeline: buffered batches times rows per batch". Neither knob bounds the delete-side allocations: GetRecords calls readAllDeleteFiles(ctx, as.fs, tasks, as.concurrency) and readAllDeletionVectors(ctx, as.fs, tasks, as.concurrency) (table/arrow_scanner.go:2179 and :2193), which materialise positional deletes and DV bitmaps for every task in the group up front, sized by delete volume rather than by batch size or buffer depth. On a delete-heavy compaction that term can dominate. Suggest narrowing the wording to the record pipeline specifically and noting the delete-side memory is not covered.
| // batch. A non-positive value keeps the table's | ||
| // read.parquet.batch-size property. | ||
| func WithCompactionReadBatchSize(n int64) CompactionGroupOption { | ||
| return func(c *compactionGroupConfig) { |
There was a problem hiding this comment.
nit — Inconsistent parameter types and names across the new option family
WithArrowBatchSize and WithCompactionReadBatchSize take int64 while WithParquetRowGroupLimit, WithRecordBatchBufferSize, WithCompactionRecordBatchBufferSize and WithCompactionParquetRowGroupLimit take int, with no evident reason for the split (the batch size is ultimately consumed via props.GetInt, which returns an int). Separately, the same underlying knob is named WithArrowBatchSize at the scan layer but WithCompactionReadBatchSize at the compaction layer, whereas the other two compaction options mirror their write-layer names exactly (WithCompaction + the write option name). Suggest aligning on int and on WithCompactionArrowBatchSize for symmetry.
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
This addresses both of the coverage gaps from the last round nicely. TestRecordBatchBufferSizeReachesRollingWriter now builds the factory through newWriterFactory and asserts cap(recordCh) end to end, so the record-buffer chain can't be silently severed anymore, and TestExecuteCompactionGroup_ParquetRowGroupLimitForwarded pins the compaction forwarding by opening the output parquet and checking the row groups are bounded. Moving WithArrowBatchSize onto a dedicated scan field also removes the WithOptions ordering trap, and I like that the WithCompactionArrowBatchSize doc now says out loud that delete-side memory isn't covered by these knobs.
Everything left is small and non-blocking: the batch-size block in arrow_scanner.go still re-clones a map that Properties() already returned fresh and guards a nil that can't happen; a couple of the doc comments describe iceberg-go-internal property keys (read.parquet.batch-size, write.parquet.row-group-limit) as portable when Java doesn't honor them; and the WithRecordBatchBufferSize doc understates peak memory on partitioned tables, where the fanout opens one channel per active partition. Details inline.
Happy to approve, will wait for merge for Matt input.
| batchSize = strconv.Itoa(as.arrowBatchSize) | ||
| } | ||
| if batchSize != "" { | ||
| tableProperties = maps.Clone(tableProperties) |
There was a problem hiding this comment.
as.metadata.Properties() already hands back a fresh clone, or iceberg.Properties{} when the map is nil, so tableProperties is never nil here and is already safe to mutate. That makes this maps.Clone a second allocation on every batch-size scan, and the if tableProperties == nil branch below it dead (it also nudges the next reader into thinking Properties() can return nil). I'd drop both and set the key in place. Minor, non-blocking.
| rec.Release() | ||
| } | ||
| assert.Equal(t, int64(numRows), totalRows) | ||
| assert.Greater(t, batches, int64(numRows/batchSize)) |
There was a problem hiding this comment.
Small robustness thing, and it predates this round: numRows/batchSize is integer division, so this really asserts batches > floor(100/7) = 14. It passes today because 100 isn't divisible by 7, but if the constants ever drift to an evenly-divisible pair (say 70 rows at 7 per batch, exactly 10 batches) then 10 > 10 is false and the test fails for the wrong reason. I'd use the ceil, assert.GreaterOrEqual(t, batches, int64((numRows+batchSize-1)/batchSize)), or just assert the exact count.
| // The cap is stored on the scan itself rather than in the options map, | ||
| // so it applies regardless of ordering relative to [WithOptions]. A | ||
| // non-positive value is ignored. | ||
| func WithArrowBatchSize(n int) ScanOption { |
There was a problem hiding this comment.
Doc nit, not blocking: read.parquet.batch-size is an iceberg-go-internal key rather than a spec property. Java uses read.parquet.vectorization.batch-size (default 5000, versus 131072 here) and PyIceberg has no read-batch-size key at all. The cap is per-scan and never persisted so nothing breaks, but a one-line note that this key doesn't match the Java convention would save someone reaching for SetProperties expecting cross-client behavior.
| // number of rows per Parquet row group in the output files. Smaller row | ||
| // groups bound the writer's buffered memory before each flush. A | ||
| // non-positive value is ignored. | ||
| func WithParquetRowGroupLimit(n int) WriteRecordOption { |
There was a problem hiding this comment.
Same interop note as WithArrowBatchSize, the other direction: write.parquet.row-group-limit is honored by iceberg-go and PyIceberg (PyIceberg defines the identical key), but Java has no row-count row-group cap. Its control is byte-based (write.parquet.row-group-size-bytes). Worth a one-line doc note so nobody expects a Java reader to respect a table that leans on this. Not blocking.
| // retained in memory until its writer consumes it, so this bound times | ||
| // the batch row count caps the memory a stalled writer can hold. The | ||
| // default is 64 batches. A non-positive value is ignored. | ||
| func WithRecordBatchBufferSize(n int) WriteRecordOption { |
There was a problem hiding this comment.
The stated memory bound holds for unpartitioned and clustered writes, but the fanout writer opens one of these channels per active partition, so peak on a partitioned table is really N_active_partitions * this bound * rows-per-batch. Might be worth saying the bound is per-partition-writer, so nobody sizes it assuming a single channel.
3 participants
Fixes #1982.
The compaction read+write pipeline had fixed memory characteristics: the scanner decodes read.parquet.batch-size rows per Arrow batch (131072 by default) with no per-scan override, and each rolling data writer buffers up to 64 record batches in its input channel. On wide rows the product of those two numbers dominates a compaction worker's peak memory.
Defaults stay exactly as they are today. Tests: table/write_read_tuning_test.go.